package com.asiainfo.www

import com.asiainfo.www.pojo.{ConfigSource, WordCount}
import org.apache.flink.api.common.functions.{FlatMapFunction, MapFunction, ReduceFunction}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSource, SingleOutputStreamOperator}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, RichCoFlatMapFunction}
import org.apache.flink.util.Collector

object SourceTestJob extends App {

  val env = StreamExecutionEnvironment.getExecutionEnvironment
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

  /**
    * confStream
    * 并行度为1广播
    */
  val configStream:DataStream[ConfigSource] = env.fromElements(
    "{\"channel\":\"APP\",\"registerDate\":\"2019-01-01\",\"historyPurchaseTimes\":0,\"maxPurchasePathLength\":1}")
    .setParallelism(1)
    .map(new MapFunction[String,ConfigSource] {
      override def map(value: String): ConfigSource = {
        ConfigSource.buildConfig(value)
      }
    })
    .broadcast()

  /**
    * 测试数据
    */
  private val test: SingleOutputStreamOperator[WordCount] = env.fromElements("test test test1 test1")
    .connect(configStream)
    .flatMap(new RichCoFlatMapFunction[String, ConfigSource, WordCount] {
      var confsource:ConfigSource = null


      override def open(parameters: Configuration): Unit = {

      }

      override def flatMap2(value: ConfigSource, out: Collector[WordCount]): Unit = {
        confsource = value
      }

      override def flatMap1(value: String, out: Collector[WordCount]): Unit = {
        for(str <- value.split(" ")){
          println("*******")
          println(confsource.toString)
          println("*******")
          out.collect(new WordCount(str, 1l))
        }
      }


    })
    .keyBy(new KeySelector[WordCount,String] {
      override def getKey(in: WordCount): String = in.word
    })
    .reduce(new ReduceFunction[WordCount] {
      override def reduce(t: WordCount, t1: WordCount): WordCount = new WordCount(t.word, t.count + t1.count)
    }).setParallelism(1)

  test.print()


  env.execute("Source test")
}
